package com.ymt.reliable.message.exceptionretry.service.impl;

import com.ymt.common.tools.DateUtils;
import com.ymt.reliable.message.exception.TransactionMessageNotExistException;
import com.ymt.reliable.message.exceptionretry.service.MessageExceptionRetryService;
import com.ymt.reliable.message.exceptionretry.task.Task;
import com.ymt.reliable.message.service.TransactionMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;

@Service
public class MessageExceptionRetryBaseOnDelayedQueue implements MessageExceptionRetryService, Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageExceptionRetryBaseOnDelayedQueue.class);
    private static final String THREDNAME = "异常消息重试主线程";
    private static DelayQueue<Task> QUEUE = new DelayQueue<>();

    @Autowired
    private TransactionMessageService transactionMessageService;

    @Override
    public void addTask(Task task) {
        synchronized (task) {
            if (!QUEUE.contains(task)) {
                QUEUE.add(task);

                LOGGER.info(">>> 任务加入本地延迟队列!taskId:{},lastPushTime:{},submitTime:{},剩余时间:{}",
                        task.getTaskId(),
                        DateUtils.formatDate(task.getMessage().getLastPushTime()),
                        DateUtils.formatDate(task.getSubmitTime()),
                        task.getDelay(TimeUnit.MILLISECONDS));
            }
        }
    }

    @Override
    public Integer queueSize() {
        return QUEUE.size();
    }

    @Override
    public void run() {
        while (true) {
            Task task = null;
            try {
                task = QUEUE.take();
                LOGGER.info(">>> 任务到期取出任务,任务编号:{},当前时间:{}", task.getTaskId(), DateUtils.currentDateTimeToString());
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                if (task != null) {
                    transactionMessageService.retry(task.getMessage().getMessageId());
                }
            } catch (TransactionMessageNotExistException e) {
                LOGGER.warn(">>> 消息不存在可能已经被消费,消息编号:{}", task.getMessage().getMessageId());
            }
        }
    }

    @PostConstruct
    public void start() {
        Thread thread = new Thread(this);
        thread.setName(THREDNAME);
        thread.start();
        LOGGER.info(">>> 异常消息重试主线程启动成功......");
    }
}
